#include "xlcomm/base/thread_pool.h"

#include <stdio.h>

#include "xlcomm/base/logger.h"

namespace xlcomm {

ThreadPool::ThreadPool(const std::string& name, int max_queue_size)
    : name_(name),
      max_queue_size_(static_cast<size_t>(max_queue_size)),
      running_(false) {
  XLCOMM_ASSERT(max_queue_size > 0);
}

ThreadPool::~ThreadPool() {
  if (running_) Stop();
}

void ThreadPool::Start(int thread_num, const Task& thread_init_callback) {
  XLCOMM_ASSERT(thread_num > 0);
  {
    std::lock_guard<std::mutex> lock(mutex_);
    if (running_) return;
    running_ = true;
  }

  LOG_INFO << "ThreadPool '" << name_
           << "' start... thread num: " << thread_num;
  thread_init_callback_ = thread_init_callback;
  threads_.reserve(static_cast<size_t>(thread_num));
  for (int i = 0; i < thread_num; ++i) {
    threads_.emplace_back(std::thread(&ThreadPool::ThreadProc, this));
  }
}

void ThreadPool::Stop() {
  {
    std::lock_guard<std::mutex> lock(mutex_);
    if (!running_) return;
    running_ = false;
    not_empty_.notify_all();
    not_full_.notify_all();
  }

  for (auto& thr : threads_) {
    thr.join();
  }
  threads_.clear();

  {
    std::lock_guard<std::mutex> lock(mutex_);
    LOG_INFO << "ThreadPool '" << name_ << "' is stopped, there is "
             << task_queue_.size() << " task discard";
    task_queue_.clear();
  }
}

void ThreadPool::AddTask(Task task) {
  if (threads_.empty()) {
    LOG_DEBUG << "ThreadPool '" << name_
              << "' has no thread now, run in caller thread\n";
    task();  // 如果没有可用线程，调用者直接执行任务
  } else {
    std::unique_lock<std::mutex> lock(mutex_);
    while (task_queue_.size() >= max_queue_size_ && running_) {
      LOG_DEBUG << "ThreadPool '" << name_ << "' task queue is full! Task "
                << &task << " wait...";
      not_full_.wait(lock);
      LOG_DEBUG << "ThreadPool '" << name_ << "' task queue is not full. Task "
                << &task << " wait finish";
    }

    if (!running_) {  // 必须在wait之后判断运行状态，防止停止后继续添加任务
      LOG_DEBUG << "ThreadPool '" << name_ << "' is not running! "
                << __FUNCTION__ << " is ignored";
      return;
    }
    task_queue_.push_back(std::move(task));
    not_empty_.notify_one();
  }
}

size_t ThreadPool::QueueSize() const {
  std::lock_guard<std::mutex> lock(mutex_);
  return task_queue_.size();
}

ThreadPool::Task ThreadPool::TakeOne() {
  std::unique_lock<std::mutex> lock(mutex_);
  while (task_queue_.empty() && running_) {
    not_empty_.wait(lock);
  }

  Task task;
  if (!task_queue_.empty()) {
    task = std::move(task_queue_.front());
    task_queue_.pop_front();
    not_full_.notify_one();
  }
  return task;
}

void ThreadPool::ThreadProc() {
  try {
    if (thread_init_callback_) {
      LOG_DEBUG << "The thread call thread_init_callback_()\n";
      thread_init_callback_();
    }
    while (running_) {
      Task task = TakeOne();
      if (task) {
        task();
      }
    }
    // } catch (const Exception& ex) {
    //   fprintf(stderr, "Exception caught in ThreadPool <%s>\n",
    //   name_.c_str()); fprintf(stderr, "Reason: %s\n", ex.what());
    //   fprintf(stderr, "Stack trace: %s\n", ex.stackTrace());
    //   abort();
  } catch (const std::exception& ex) {
    fprintf(stderr, "Exception caught in ThreadPool '%s'\n", name_.c_str());
    fprintf(stderr, "Reason: %s\n", ex.what());
    abort();
  } catch (...) {
    fprintf(stderr, "Unknown exception caught in ThreadPool '%s'\n",
            name_.c_str());
    throw;  // rethrow
  }
}

}  // namespace xlcomm